package com.zeng.ssm.service.support.async;

import cn.hutool.core.net.NetUtil;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.zeng.ssm.api.support.idempotent.DistributedLockService;
import com.zeng.ssm.dal.mapper.AsyncTaskConfigDao;
import com.zeng.ssm.dal.mapper.AsyncTaskRecordDao;
import com.zeng.ssm.dal.model.AsyncTaskConfig;
import com.zeng.ssm.dal.model.AsyncTaskRecord;
import com.zeng.ssm.service.support.async.dto.AsyncTaskConfigDto;
import com.zeng.ssm.service.support.async.dto.AsyncTaskRecordDto;
import com.zeng.ssm.service.util.ApplicationContextUtil;
import com.zeng.ssm.util.DateUtil;
import com.zeng.ssm.util.JsonUtil;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.helpers.MessageFormatter;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;

import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 异步任务-线程池+时间轮
 * <p>
 * 通过数据库乐观锁抢占任务
 * <p>
 * 定时执行锁续期任务，为已被抢占的任务锁续期
 * <p>
 * 定时执行锁超时移除任务，防止因服务宕机等原因造成任务死锁
 * <p>
 * 一个任务被加入时间轮后，如果执行失败，再次加入时间轮，直到成功/最终失败/异常才释放任务
 *
 * @author zengwanlin
 * @date 2023-03-31T13:29
 */
@Component
public class AsyncTaskExecutorLauncher implements InitializingBean, DisposableBean {

    private Logger logger = LoggerFactory.getLogger(AsyncTaskExecutorLauncher.class);

    @Autowired
    private AsyncTaskConfigDao asyncTaskConfigDao;

    @Autowired
    private AsyncTaskRecordDao asyncTaskRecordDao;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private DistributedLockService distributedLockService;

    @Autowired
    private AsyncTaskConfigurer configurer;

    private HashedWheelTimer hashedWheelTimer;

    private ScheduledThreadPoolExecutor threadPoolExecutor;

    public static final String LOCAL_IP4_ADDRESS = StringUtils.defaultIfEmpty(NetUtil.getLocalhostStr(), "127.0.0.1");

    /**
     * 锁持有者
     */
    private static final String LOCK_OWNER = LOCAL_IP4_ADDRESS;

    /**
     * 锁持有者-未被持有时为NONE
     */
    public static final String DEFAULT_LOCK_OWNER = "NONE";

    /**
     * 锁持有期限（s）
     */
    public static final Integer DEFAULT_LOCK_TTL = 30;

    /**
     * 抢占到锁的任务数据
     * <p>
     * 用于锁续期
     */
    private final String GRAB_ASYNC_TASK_KEY = "GRAB_ASYNC_TASK:" + LOCK_OWNER;

    @Override
    public void afterPropertiesSet() throws Exception {
        // 初始化线程池
        List<AsyncTaskConfig> taskConfigList = asyncTaskConfigDao.selectValidRecord();
        if (CollectionUtils.isEmpty(taskConfigList)) {
            logger.info("无异步任务，不初始化AsyncTask");
            return;
        }
        logger.info("初始化异步任务");
        LoggingUncaughtExceptionHandler exceptionHandler = new LoggingUncaughtExceptionHandler();
        int corePoolSize = Math.min(taskConfigList.size() + 3, Runtime.getRuntime().availableProcessors() * 2);
        ThreadFactoryBuilder taskFactoryBuilder = new ThreadFactoryBuilder();
        taskFactoryBuilder.setNameFormat("AsyncTask-%d").setUncaughtExceptionHandler(exceptionHandler);
        if (this.configurer.getRejectedExecutionHandler() == null) {
            this.configurer.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        }
        ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(corePoolSize, taskFactoryBuilder.build(), this.configurer.getRejectedExecutionHandler());
        executor.setMaximumPoolSize(this.configurer.getMaxPoolSize());
        executor.setKeepAliveTime(this.configurer.getKeepAliveTime(), TimeUnit.SECONDS);
        executor.setContinueExistingPeriodicTasksAfterShutdownPolicy(false);
        executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
        this.threadPoolExecutor = executor;

        // 时间轮
        ThreadFactoryBuilder wheelFactoryBuilder = new ThreadFactoryBuilder();
        wheelFactoryBuilder.setNameFormat("AsyncTaskWheelTimer-%d").setUncaughtExceptionHandler(exceptionHandler);
        this.hashedWheelTimer = new HashedWheelTimer(wheelFactoryBuilder.build(), 1L, TimeUnit.SECONDS, 60);

        // 移除已存在任务数据
        redisTemplate.delete(GRAB_ASYNC_TASK_KEY);

        // 抢占任务
        scheduleGrabTask();

        // 锁续期任务
        scheduleLockRenewal();

        // 锁超时任务
        scheduleLockExpire();
    }

    @Override
    public void destroy() throws Exception {
        if (this.hashedWheelTimer != null) {
            logger.info("停止时间轮");
            this.hashedWheelTimer.stop();
        }
        if (this.threadPoolExecutor != null) {
            logger.info("关闭线程池");
            this.threadPoolExecutor.shutdown();
        }
        if (this.redisTemplate != null) {
            logger.info("移除已抢占任务缓存");
            redisTemplate.delete(GRAB_ASYNC_TASK_KEY);
        }
    }

    /**
     * 抢占任务
     */
    private void scheduleGrabTask() {
        this.threadPoolExecutor.scheduleAtFixedRate(ThreadTraceNoUtil.wrap(new Runnable() {
            @Override
            public void run() {
                if (logger.isTraceEnabled()) {
                    logger.trace("[{}]执行抢占任务", LOCK_OWNER);
                }
                List<AsyncTaskRecord> taskRecordList = asyncTaskRecordDao.selectGrabableRecordList();
                AsyncTaskConfig taskConfig;
                for (AsyncTaskRecord taskRecord : taskRecordList) {
                    if (doGrabTask(taskRecord.getId())) {
                        logger.info("服务[{}]抢占到任务，ID[{}]，名称[{}]", new Object[]{LOCK_OWNER, taskRecord.getId(), taskRecord.getTaskCode() + AsyncTaskConstant.HORIZONTAL_LINE + taskRecord.getTaskName()});
                        taskConfig = asyncTaskConfigDao.selectByTaskCode(taskRecord.getTaskCode());
                        // 提交超时任务到时间轮
                        hashedWheelTimer.newTimeout(createTimerTask(taskConfig, taskRecord), taskConfig.getIntervalTime().longValue(), TimeUnit.SECONDS);
                        // 缓存已抢占成功的任务
                        cacheGrabTask(taskRecord.getId());
                    }
                }
            }
        }), this.configurer.getInitialDelay(), this.configurer.getGrabTaskInterval(), TimeUnit.SECONDS);
    }

    private class LoggingUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {
        @Override
        public void uncaughtException(Thread t, Throwable e) {
            logger.error(MessageFormatter.arrayFormat("线程ID[{}]，线程名[{}]执行出错，异常信息：[{}]", new String[]{Long.toString(t.getId()), t.getName(), e.getMessage()}).getMessage(), e);
        }
    }

    /**
     * 抢占任务
     *
     * @param recordId
     * @return
     */
    private boolean doGrabTask(Integer recordId) {
        AsyncTaskRecord record = new AsyncTaskRecord();
        record.setLockOwner(LOCK_OWNER);
        record.setUpdateTime(DateUtil.getDateTime());

        AsyncTaskRecord example = new AsyncTaskRecord();
        example.setId(recordId);
        example.setLockOwner(DEFAULT_LOCK_OWNER);

        int rows = asyncTaskRecordDao.updateByExample(record, example);
        return rows == 1;
    }

    /**
     * 注册锁续期任务-已抢占锁成功后给锁续期
     */
    private void scheduleLockRenewal() {
        if (this.configurer.getLockRenewalInterval() >= DEFAULT_LOCK_TTL - 5) {
            throw new IllegalArgumentException("lockRenewalInterval值必须小于" + (DEFAULT_LOCK_TTL - 5));
        }
        this.threadPoolExecutor.scheduleAtFixedRate(ThreadTraceNoUtil.wrap(new Runnable() {
            @Override
            public void run() {
                if (logger.isTraceEnabled()) {
                    logger.trace("[{}]执行锁续期任务", LOCK_OWNER);
                }
                Set<Integer> idSet = redisTemplate.opsForSet().members(GRAB_ASYNC_TASK_KEY);
                if (CollectionUtils.isEmpty(idSet)) {
                    return;
                }
                AsyncTaskRecord record;
                AsyncTaskRecord taskRecord;
                for (Integer id : idSet) {
                    taskRecord = asyncTaskRecordDao.selectByPrimaryKey(id);
                    if (!LOCK_OWNER.equals(taskRecord.getLockOwner())) {
                        // 移除其他拥有者的任务
                        removeCachedGrabTask(id);
                        continue;
                    }
                    record = new AsyncTaskRecord();
                    record.setId(id);
                    record.setUpdateTime(DateUtil.getDateTime());
                    int row = asyncTaskRecordDao.updateByPrimaryKeySelective(record);
                    logger.debug("异步任务[{}]锁续期，影响行[{}]", id, row);
                }
            }
        }), this.configurer.getInitialDelay(), this.configurer.getLockRenewalInterval(), TimeUnit.SECONDS);
    }

    /**
     * 注册锁过期任务-已抢占任务锁失效任务
     */
    private void scheduleLockExpire() {
        this.threadPoolExecutor.scheduleAtFixedRate(ThreadTraceNoUtil.wrap(new Runnable() {
            @Override
            public void run() {
                final String lockKey = "ASYNC_TASK_LOCK_EXPIRE";
                boolean lockSuc = distributedLockService.lock(lockKey, lockKey, configurer.getLockExpireInterval());
                if (!lockSuc) {
                    logger.debug("锁释放任务，其他服务执行中，跳过");
                    return;
                }
                try {
                    List<AsyncTaskRecord> taskRecordList = asyncTaskRecordDao.selectLockedRecord();
                    AsyncTaskRecord record = null;
                    for (AsyncTaskRecord item : taskRecordList) {
                        if (hasExpired(item.getUpdateTime(), item.getLockTtl())) {
                            logger.info("任务[{}]已过期", item.getId());
                            record = new AsyncTaskRecord();
                            record.setId(item.getId());
                            record.setLockOwner(DEFAULT_LOCK_OWNER);
                            record.setUpdateTime(DateUtil.getDateTime());
                            asyncTaskRecordDao.updateByPrimaryKeySelective(record);
                        }
                    }
                } finally {
                    distributedLockService.unlock(lockKey, lockKey);
                }
            }
        }), this.configurer.getInitialDelay(), this.configurer.getLockExpireInterval(), TimeUnit.SECONDS);
    }

    /**
     * 判断任务占有锁是否已过期
     *
     * @param dateTimeStr
     * @param ttl
     * @return true-已过期，false-未过期
     */
    private boolean hasExpired(String dateTimeStr, long ttl) {
        Date lockDate = DateUtil.dateTimeToDate(dateTimeStr);
        long diff = TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis() - lockDate.getTime());
        return diff >= ttl;
    }

    private TimerTask createTimerTask(final AsyncTaskConfig taskConfig, final AsyncTaskRecord taskRecord) {
        return new TimerTask() {
            @Override
            public void run(Timeout timeout) throws Exception {
                threadPoolExecutor.execute(ThreadTraceNoUtil.wrap(new Runnable() {
                    @Override
                    public void run() {
                        boolean removeCache = true;
                        try {
                            BaseAsyncTaskExecutor executor = ApplicationContextUtil.getBean(taskConfig.getExecutorId(), BaseAsyncTaskExecutor.class);
                            Class<?> paramClass = loadClass(taskConfig.getParamClassName());
                            Object param = JsonUtil.fromJson(taskRecord.getParamJson(), paramClass);

                            AsyncTaskContext context = new AsyncTaskContext();
                            context.setParam(param);
                            context.setRetMsg("Success");
                            AsyncTaskConfigDto taskConfigDto = new AsyncTaskConfigDto();
                            AsyncTaskRecordDto taskRecordDto = new AsyncTaskRecordDto();
                            BeanUtils.copyProperties(taskConfig, taskConfigDto);
                            BeanUtils.copyProperties(taskRecord, taskRecordDto);
                            context.setTaskConfigDto(taskConfigDto);
                            context.setTaskRecordDto(taskRecordDto);

                            if (!executor.succeed(context)) {
                                // 执行任务
                                executor.execute(context);

                                if (AsyncTaskConstant.ASYNC_TASK_STATUS_2.equals(context.getStatus())) {
                                    // 重新添加到时间轮
                                    AsyncTaskConfig asyncTaskConfig = asyncTaskConfigDao.selectByPrimaryKey(taskConfig.getId());
                                    AsyncTaskRecord asyncTaskRecord = asyncTaskRecordDao.selectByPrimaryKey(taskRecord.getId());
                                    if (AsyncTaskConstant.VALID_FLAG_1.equals(asyncTaskConfig.getValidFlag()) && asyncTaskRecord.getExecuteTimes() < asyncTaskConfig.getMaxExecuteTimes()) {
                                        logger.info("异步任务ID[{}] taskCode[{}]执行失败，重新加入时间轮", asyncTaskRecord.getId(), asyncTaskRecord.getTaskCode());
                                        hashedWheelTimer.newTimeout(createTimerTask(asyncTaskConfig, asyncTaskRecord), asyncTaskConfig.getIntervalTime().longValue(), TimeUnit.SECONDS);
                                        removeCache = false;
                                    } else {
                                        logger.info("异步任务[{}]不满足执行条件，终止处理", asyncTaskConfig.getTaskCode());
                                    }
                                }
                            } else {
                                executor.alreadySucceed(context);
                            }
                        } catch (Throwable th) {
                            logger.error(th.getMessage(), th);
                        } finally {
                            if (removeCache) {
                                // 移除数据库中的锁
                                asyncTaskRecordDao.unlock(taskRecord.getId(), LOCK_OWNER);
                                // 移除已抢占的任务
                                removeCachedGrabTask(taskRecord.getId());
                            }
                        }
                    }
                }));
            }
        };
    }

    private Class<?> loadClass(String className) {
        try {
            return ClassUtils.forName(className, ClassUtils.getDefaultClassLoader());
        } catch (ClassNotFoundException e) {
            throw new IllegalArgumentException(e);
        }
    }

    /**
     * 添加抢占到的任务
     *
     * @param id ASYNC_TASK_RECORD.ID
     */
    private void cacheGrabTask(Integer id) {
        redisTemplate.opsForSet().add(GRAB_ASYNC_TASK_KEY, id);
    }

    /**
     * 移除抢占到的任务
     *
     * @param id ASYNC_TASK_RECORD.ID
     */
    private void removeCachedGrabTask(final Integer id) {
        redisTemplate.opsForSet().remove(GRAB_ASYNC_TASK_KEY, id);
    }

}
